%%%====================================================================
%%% File    : pipe_disk_logger.erl
%%% Author  :  <vjache>
%%% Description: Pipe to write terms to internally formatted binary log
%%%				 using kernel module 'disk_log'.
%%%
%%% Created : 24.03.2009
%%%====================================================================

-module(pipe_disk_logger).

%%
%% Include files
%%

-include_lib("eunit/include/eunit.hrl").

-include("../include/pipe.hrl").
%%
%% Exported Functions
%%
-export([
		 new/1,
		 new/2,
		 new/3,
		 new/4,
		 open/1,
		 push/2,
		 flush/1,
		 close/1,
		 test/1,
		 test1/1,
		 test2/1,
		 test_gendata/1
		 ]).

-record(state,{log,
			   next_pipe_state,
			   buff,
			   buff_capacity,
			   buff_free_size,
			   close_fun=undefined}).

-define(CHECKERR(Expr,ErrTag),checkerr(Expr,ErrTag)).
-define(GUARD_STATE,is_record(State,state)).

%%====================================================================
%% API Functions
%%====================================================================

new(ArgL) ->
	new(ArgL,pipe:terminator()).

new(ArgL,NextPipe) ->
	new(ArgL,1000,NextPipe).
new(ArgL,BuffSize,NextPipe) 
  when is_list(ArgL),
	   is_integer(BuffSize),
	   is_record(NextPipe, pipe),
	   BuffSize>0 ->
	pipe:pipe_from_module(?MODULE, {ArgL,BuffSize}, NextPipe);
new(BuffSize,OpenFun,CloseFun)
  when is_function(OpenFun, 0),
	   is_function(CloseFun, 0),
	   is_integer(BuffSize),
	   BuffSize>0 ->
	new(BuffSize,OpenFun,CloseFun,pipe:terminator()).
new(BuffSize,OpenFun,CloseFun,NextPipe) 
  when is_function(OpenFun, 0),
	   is_function(CloseFun, 0),  
	   is_integer(BuffSize),
	   is_record(NextPipe, pipe),
	   BuffSize>0 ->
	pipe:pipe_from_module(?MODULE, {BuffSize,OpenFun,CloseFun}, NextPipe).

open({{ArgL,BuffSize},NextPipe}) -> %%PipeState
	case lists:keysearch(name, 1, ArgL) of
		false ->
			throw({name_absent,ArgL});
		{value,{name,Log}} ->
			?CHECKERR(disk_log:open(ArgL),open_err),
			#state{
				   log=Log,
				   next_pipe_state=pipe:open(NextPipe),
				   buff=[],
				   buff_capacity=BuffSize,
				   buff_free_size=BuffSize}
	end;
open({{BuffSize,OpenFun,CloseFun},NextPipe}) -> %%PipeState
	Log=?CHECKERR(OpenFun(),open_err),
	#state{
		   log=Log,
		   next_pipe_state=pipe:open(NextPipe),
		   buff=[],
		   buff_capacity=BuffSize,
		   buff_free_size=BuffSize,
		   close_fun=CloseFun}.

push(State,?EOF) when ?GUARD_STATE ->
	State;
push(State,Item) when 
  ?GUARD_STATE, 
  State#state.buff_free_size==0 ->
	push(flush_buff(State),Item);
push(State,Item) when 
  ?GUARD_STATE, 
  State#state.buff_free_size>0 ->
	NextPipeState=State#state.next_pipe_state,
	State#state{
				buff=[Item|State#state.buff],
				buff_free_size=State#state.buff_free_size-1,
				next_pipe_state=pipe:push(NextPipeState,Item)
				}.
flush(State) when ?GUARD_STATE ->
	State1=flush_buff(State),
	Log=State#state.log,
	?CHECKERR(disk_log:sync(Log),sync_err),
	State1.
close(State) when ?GUARD_STATE ->
	Log=State#state.log,
	NextPipeState=State#state.next_pipe_state,
	case State#state.close_fun of
		undefined ->
			?CHECKERR(disk_log:close(Log),close_err);
		CloseFun ->
			?CHECKERR(CloseFun(),close_err)
	end,
	pipe:close(NextPipeState),
	ok.

flush_buff(State) when ?GUARD_STATE ->
	Log=State#state.log,
	?CHECKERR(disk_log:log_terms(Log, lists:reverse(State#state.buff)),log_terms_err),
	State#state{
				buff=[],
				buff_free_size=State#state.buff_capacity
			   }.

%%====================================================================
%% Test Functions
%%====================================================================

test_gendata(N) ->
	{A1,A2,A3} = now(),
    random:seed(A1, A2, A3),
	PipeState=pipe:open(new([{name,"BigData"}])),
	test_gendata(N,PipeState,0),
	pipe:close(PipeState).

test_gendata(-1,_PipeState,_Cnt) ->
	ok;
test_gendata(N,PipeState,Cnt) ->
	case N rem 10000 of
		0 ->
			io:format("~p tuples to insert~n",[N]);
		_ ->
			ok
	end,
	Id=random:uniform(1000),
	NewPipeState=pipe:push(PipeState,{Id,"Some meaning value #"++integer_to_list(Cnt)}),
	test_gendata(N-1,NewPipeState,Cnt+1).


test(L) when is_list(L) ->
	S=pipe:open(new([{name,"TestLog"}])),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).
test1(L) when is_list(L) ->
	S=pipe:open(new(pipe_console:new())),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).
test2(L) when is_list(L) ->
	S=pipe:open(new("c:/pipe_logger.txt",pipe_console:new())),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).


%%====================================================================
%% Local Functions
%%====================================================================

checkerr({error,Reason},ErrTag) ->
	throw({ErrTag,Reason});
checkerr(Result,_ErrTag) ->
	Result.

%%====================================================================
%% Test Functions
%%====================================================================
basic_test_() ->
	{foreach, 
	 local, 
	 fun setup_/0, 
	 fun teardown_/1, 
	 [
	  fun(ArgL)->
			  ?_test(test_open_close(ArgL))
	  end,
	  fun(ArgL)->
			  ?_test(test_write(ArgL))
	  end
%% 	 ,
%% 	  fun(ArgL)->
%% 			  ?_test(test_open_close_c(ArgL))
%% 	  end,
%% 	  fun(ArgL)->
%% 			  ?_test(test_write_c(ArgL))
%% 	  end
	 ]}.

setup_() ->
	Filename="./pipe_test.log",
	case filelib:is_regular(Filename) of
		true ->
			?CHECKERR(file:delete(Filename),del_file_err);
		false ->
			ok
	end,
	[{name,{rw,Filename}},{file,Filename}].

teardown_(ArgL) ->
	case lists:keysearch(file, 1, ArgL) of
		{value, {file,Filename}} ->
			?CHECKERR(file:delete(Filename),del_file_err);
		false ->
			throw(unexpected_branch)
	end.

test_open_close(ArgL) ->
	Pipe=new(ArgL),
	pipe:close(pipe:open(Pipe)).
test_write(ArgL) ->
	Pipe=new(ArgL),
	PipeState0=pipe:open(Pipe),
	PipeState1=lists:foldl(fun(Item,PipeState)->
						pipe:push(PipeState,Item)
				end, PipeState0, lists:seq(1, 5005)),
	pipe:close(PipeState1).
%% test_open_close_c(ArgL) ->
%% 	{value, {name,Log}}=lists:keysearch(name, 1, ArgL),
%% 	Pipe=new(100,
%% 			 fun()->
%% 					 
%% 					 disk_log_pool:open_log(disk_log_pool,Log),
%% 					 Log
%% 			 end,
%% 			 fun()->
%% 					 disk_log_pool:close_finalize_log(disk_log_pool,Log)
%% 			 end),
%% 	pipe:close(pipe:open(Pipe)).
%% test_write_c(ArgL) ->
%% 	{value, {name,Log}}=lists:keysearch(name, 1, ArgL),
%% 	Pipe=new(100,
%% 			 fun()->
%% 					 
%% 					 disk_log_pool:open_log(disk_log_pool,Log),
%% 					 Log
%% 			 end,
%% 			 fun()->
%% 					 disk_log_pool:close_finalize_log(disk_log_pool,Log)
%% 			 end),
%% 	PipeState0=pipe:open(Pipe),
%% 	PipeState1=lists:foldl(fun(Item,PipeState)->
%% 						pipe:push(PipeState,Item)
%% 				end, PipeState0, lists:seq(1, 5005)),
%% 	pipe:close(PipeState1).